using CoreManager.MessageQueue.Models;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;

namespace CoreManager.MessageQueue.Services;

/// <summary>
/// Kafka后台服务，用于自动启动消费者
/// </summary>
public class KafkaBackgroundService : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger<KafkaBackgroundService> _logger;
    private readonly KafkaOptions _options;

    public KafkaBackgroundService(
        IServiceProvider serviceProvider, 
        ILogger<KafkaBackgroundService> logger,
        IOptions<KafkaOptions> options)
    {
        _serviceProvider = serviceProvider;
        _logger = logger;
        _options = options.Value;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Kafka background service starting");

        try
        {
            using var scope = _serviceProvider.CreateScope();
            var consumerService = scope.ServiceProvider.GetRequiredService<IKafkaConsumerService>();

            // 从配置中获取所有订阅主题
            var topics = _options.Topics.GetAllSubscriptionTopics();
            
            if (topics.Any())
            {
                _logger.LogInformation("Subscribing to topics: {Topics}", string.Join(", ", topics));
                consumerService.Subscribe(topics);
            }
            else
            {
                _logger.LogWarning("No topics configured for subscription");
                return;
            }

            // 启动消费
            await consumerService.StartConsumingAsync(HandleMessageAsync, stoppingToken);
        }
        catch (OperationCanceledException)
        {
            _logger.LogInformation("Kafka background service was cancelled");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error in Kafka background service");
        }
        finally
        {
            _logger.LogInformation("Kafka background service stopped");
        }
    }

    /// <summary>
    /// 处理接收到的消息
    /// </summary>
    /// <param name="topic">主题</param>
    /// <param name="message">消息</param>
    /// <returns>处理是否成功</returns>
    private async Task<bool> HandleMessageAsync(string topic, KafkaMessage message)
    {
        try
        {
            _logger.LogInformation("Processing message {MessageId} from topic {Topic}, type: {MessageType}", 
                message.MessageId, topic, message.MessageType);

            // 根据消息类型处理不同的业务逻辑
            switch (message.MessageType)
            {
                case "UserLogin":
                    await HandleUserLoginMessage(message);
                    break;
                case "SystemAlert":
                    await HandleSystemAlertMessage(message);
                    break;
                case "DataSync":
                    await HandleDataSyncMessage(message);
                    break;
                default:
                    _logger.LogWarning("Unknown message type: {MessageType}", message.MessageType);
                    break;
            }

            return true;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to process message {MessageId}", message.MessageId);
            return false;
        }
    }

    private async Task HandleUserLoginMessage(KafkaMessage message)
    {
        // 处理用户登录消息
        _logger.LogInformation("Processing user login message: {Data}", message.Data);
        await Task.Delay(100); // 模拟处理时间
    }

    private async Task HandleSystemAlertMessage(KafkaMessage message)
    {
        // 处理系统告警消息
        _logger.LogInformation("Processing system alert message: {Data}", message.Data);
        await Task.Delay(100); // 模拟处理时间
    }

    private async Task HandleDataSyncMessage(KafkaMessage message)
    {
        // 处理数据同步消息
        _logger.LogInformation("Processing data sync message: {Data}", message.Data);
        await Task.Delay(100); // 模拟处理时间
    }
}
